package dbms;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileWriter;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Hashtable;
import java.util.Iterator;

public class RessourcenManager {
	
	private static final int PUFFER_SIZE = 0;
	private final int _rmid;
	
	
	public RessourcenManager(int rmid) {
		_rmid = rmid;
	}
	
	//offene SubTransaktionen
	private Hashtable<Integer, SubTransaction> _currentTAs = new Hashtable<Integer, SubTransaction>();
	
	//buffer
	private Hashtable<Integer, String> _puffer = new Hashtable<Integer, String>();
	
	public int getID() {
		return _rmid;
	}
	
	/**
	 * Beginnt neue SubTransaktion
	 * @param clientID
	 * @return
	 */
	public void beginSubTransaction(int taid) {
		
		SubTransaction subta = new SubTransaction(taid);
		_currentTAs.put(subta.getTaID(), subta);
		
		Koordinator.getInstance().reg(this, taid);

		this.log("[" + this.getNextLSN() + "," + "BOT" + "," + subta.getTaID() + "]" );
	}
	
	/**
	 * Beendet SubTransaktion
	 * @param taid
	 */
	public boolean commit(int taid) {

		this.log("[" + this.getNextLSN() + "," + "COMMIT" + "," +taid+ "]");
		_puffer.putAll(_currentTAs.get(taid).getWork());
		_currentTAs.remove(taid);

		if ((_puffer.size()) > PUFFER_SIZE) 
			this.writePuffer();	
		
		return true;
	}
	
	/**
	 * bereitet commit vor
	 * @param taid
	 * @return
	 */
	public boolean prepare(int taid) {
		
		boolean prepared = true;
		
		this.log("[" + this.getNextLSN() + ",PREPARED,"+taid+"]");
		
		return prepared;
	}
	
	/**
	 * Löscht writebefehle aus dem Puffer
	 * @param taid
	 * @return
	 */
	public boolean rollback(int taid) {
		
		_currentTAs.get(taid).clearWork();
		
		this.log("[" + this.getNextLSN() + ",ROLLBACK,"+taid+"]");
		
		return true;
	}
	
	/**
	 * Schreibt Datensatz in den Puffer
	 * @param taid
	 * @param pageid
	 * @param data
	 */
	public synchronized boolean write(int taid, int pageid, String data) {
		
		if (!_currentTAs.containsKey(taid)) {
			this.beginSubTransaction(taid);
		}
		
		int lsn = this.getNextLSN();
		
		this.log("["+ lsn + "," + taid+ "," + pageid+ "," + data + "]");
		
		_currentTAs.get(taid).addWork(pageid, "["+ pageid + "," + lsn + "," + data + "]");
		
		return true;
	}
	
	/**
	 * Speichert den Puffer auf Persistenz
	 */
	public synchronized void writePuffer() {

		 Iterator<Integer> it = _puffer.keySet().iterator();
		 
         while(it.hasNext()) {
        	 
        	int pageid = Integer.valueOf(it.next().toString());           
 
			try {
				
				//Datei erstellen, falls nicht vorhanden
				File f = new File("data/" + pageid + ".txt");
				if(!f.exists())
					f.createNewFile();
				
				//Daten in Seite(Datei) schreiben
				FileWriter fw = new FileWriter("data/" + pageid + ".txt");
				fw.write(_puffer.get(pageid).toString());
				fw.close();

			} catch (FileNotFoundException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			}
         }

         _puffer.clear();
	}
	
	/**
	 * Schreibt Logeintrag in Log
	 * @param logEntry
	 */
	public synchronized void log(String logEntry){
		
		try {
			
			//Datei erstellen, falls nicht vorhanden
			File f = new File("data/log_rm" + _rmid + ".txt");
			if(!f.exists())
				f.createNewFile();
			
			//Daten in Seite(Datei) schreiben
			FileWriter fw = new FileWriter("data/log_rm" + _rmid + ".txt",true);
			fw.write(logEntry + System.getProperty("line.separator"));
			fw.close();
			
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		System.out.println(logEntry);
	}
	
	/**
	 * Führt, wenn nötig, Recovery mittels Logdaten aus
	 */
	public void recover() {
		
		Collection<Integer> winner = new ArrayList<Integer>();
		Collection<Integer> prepared = new ArrayList<Integer>();
		Collection<Integer> aborted = new ArrayList<Integer>();
		String logLine;
		
		try {
		       
			RandomAccessFile logFile = new RandomAccessFile("data/log_rm" + _rmid + ".txt", "rw");
				
			   // sammelt alle Gewinnertransaktionen	
		       while( (logLine = logFile.readLine()) != null ) {

		    	   if (logLine.contains("COMMIT")) {
		    		   
		    		   String taid = logLine.substring(logLine.indexOf("COMMIT") + 7, logLine.indexOf("]"));

		    		   winner.add(Integer.valueOf(taid));
		    	   }
		    	   
		    	   else if (logLine.contains("ROLLBACK")) {
		    		   
		    		   String taid = logLine.substring(logLine.indexOf("ROLLBACK") + 9, logLine.indexOf("]"));

		    		   aborted.add(Integer.valueOf(taid));
		    	   }
		    	   
		    	   else if (logLine.contains("PREPARED")) {
		    		   
		    		   String taid = logLine.substring(logLine.indexOf("PREPARED") + 9, logLine.indexOf("]"));

		    		   prepared.add(Integer.valueOf(taid));
		    	   }
		       }
		       
		       prepared.removeAll(aborted);
		       
		       // wieder zum Dateianfang zurück
		       logFile.seek(0);
		       
		       // alle Writebefehle der Gewinner durchgehen und mit Seiteninhalt vergleichen, dann ggf. redo 
		       while( (logLine = logFile.readLine()) != null ) { 
		    	   
		    	   if (!logLine.contains("COMMIT") && !logLine.contains("BOT") && !logLine.contains("PREPARED") && !logLine.contains("ROLLBACK")) {
		    		   
		    		   String str = logLine.substring(logLine.indexOf(",")+1);
		    		   String taid = str.substring(0, str.indexOf(","));
		    		   String lsn = logLine.substring(1, logLine.indexOf(","));
	    			   String pageid = str.substring(str.indexOf(",")+1, str.lastIndexOf(","));
	    			   String data = str.substring(str.lastIndexOf(",")+1, str.indexOf("]"));

		    		   if (winner.contains(Integer.valueOf(taid))) {

		    			   RandomAccessFile site = new RandomAccessFile("data/"+pageid+".txt", "rw");
		    			   
		    			   String siteLine = site.readLine(); 
		    			   
		    			   if (siteLine != null) {
		    			   
			    			   int siteLSN = Integer.valueOf(siteLine.substring(siteLine.indexOf(",")+1, siteLine.lastIndexOf(",")));
			    			   
			    			   if (siteLSN < Integer.valueOf(lsn)) {

			    				   site.seek(0);
			    				   
			    				   site.write(("["+ pageid + "," + lsn + "," + data + "]").getBytes());
			    				   
			    				   System.out.println("[REDO,"+ pageid + "," + lsn + "," + data + "]");
			    			   }
		    			   }

		    			   else {
		    				   site.write(("["+ pageid + "," + lsn + "," + data + "]").getBytes());
		    				   System.out.println("[REDO,"+ pageid + "," + lsn + "," + data + "]");
		    			   }
		    			   
		    			   site.close();
		    		   }
		    		   
		    		   else if (prepared.contains(Integer.valueOf(taid))) {
	    				   
	    				   if (!_currentTAs.contains(Integer.valueOf(taid))) {
	    					   
	    					   SubTransaction subta = new SubTransaction(Integer.valueOf(taid));
	    					   
	    					   subta.setPrepareStatus(true);
	    					   
	    						_currentTAs.put(subta.getTaID(), subta);
	    						
	    						Koordinator.getInstance().reg(this, Integer.valueOf(taid));
	    				   }
	    				   
	    				   _currentTAs.get(Integer.valueOf(taid)).addWork(Integer.valueOf(pageid), "["+ pageid + "," + lsn + "," + data + "]");
	    				   
	    				   System.out.println("[REDO_PREPARE,"+ pageid + "," + lsn + "," + data + "]");
	    				   
	    			   }
		    	   } 
		       }
		       
		       logFile.close();
		       
		       for (int id : prepared) {
		    	   
		    	   if (Koordinator.getInstance().isCommitted(id)) {
		    		   this.commit(id);
		    	   }
		    	   else {
		    		   this.rollback(id);
		    	   }
		       }

   
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * Gibt nächste LSN zurück und speichert diese im Config-File
	 * @return
	 */
	public synchronized int getNextLSN() {
		
		int newID = 0;
		
		try {
			
			RandomAccessFile config = new RandomAccessFile("data/config_rm"+ _rmid +".txt", "rw");
			
			String configLine = config.readLine(); 
			
			if (configLine != null) {
				
				newID = Integer.valueOf(configLine.substring(configLine.indexOf(",")+1))+1;
				
				configLine = configLine.substring(0,configLine.indexOf(",")+1);
				configLine = configLine + newID;
				config.seek(0);
				config.write(configLine.getBytes());
			
			}
			
			else {
				config.write("0,0".getBytes());
			}
			
			config.close();
			 
		} catch (FileNotFoundException e) {
			e.printStackTrace();
			
		} catch (IOException e) {
			e.printStackTrace();
		}

		return newID;
	}
}
